import io.vertx.core.AbstractVerticle
import io.vertx.core.Vertx
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently
import org.apache.rocketmq.client.producer.DefaultMQProducer
import org.apache.rocketmq.common.consumer.ConsumeFromWhere
import org.apache.rocketmq.common.message.Message

class RocketMQSenderVerticle : AbstractVerticle() {
  override fun start() {
    val producer = DefaultMQProducer("Msg1")
    producer.namesrvAddr = "8.130.168.253:9876"
    producer.start()

    vertx.setPeriodic(1000) {
      val message = Message("YourTopic", "YourTags", "Hello, RocketMQ from Vert.x".toByteArray())
      producer.sendOneway(message)
      println("Message sent to RocketMQ")
    }
  }
}

class RocketMQReceiverVerticle : AbstractVerticle() {
  override fun start() {
    val consumer = DefaultMQPushConsumer("Msg1")
    consumer.namesrvAddr = "8.130.168.253:9876"
    consumer.consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
    consumer.subscribe("YourTopic", "YourTags")

    consumer.registerMessageListener(MessageListenerConcurrently { msgs, context ->
      for (msg in msgs) {
        val message = String(msg.body)
        println("产生消息"+message)
        // 处理消息
        vertx.eventBus().send("rocketmq_message_received", message)
      }
      ConsumeConcurrentlyStatus.CONSUME_SUCCESS
    })

    consumer.start()
  }
}

fun main() {
  val vertx = Vertx.vertx()

  vertx.deployVerticle(RocketMQSenderVerticle())
  vertx.deployVerticle(RocketMQReceiverVerticle())
}
